package net.flyinggroup.kafka.domaincrawler;


import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

@Service
@Slf4j
public class DomainCrawlerService {
    private final KafkaTemplate<String, Domain> kafkaTemplate;
    private final String KAFKA_TOPIC = "web-domains";

    public DomainCrawlerService(KafkaTemplate<String, Domain> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void crawl(String name) {
        Mono<DomainList> domainListMono = WebClient.create()
                .get()
                .uri(String.format("https://api.domainsdb.info/v1/domains/search?domain=%s&zone=com", name))
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToMono(DomainList.class);
        domainListMono.subscribe(domainList -> {
            domainList.domains.forEach(domain -> {
                kafkaTemplate.send(KAFKA_TOPIC, domain);
                log.info("Domain message: {} ", domain.getDomain());
            });
        });
    }
}
